package com.yugao.fintech.antelope.mps.core.propcess;

import com.alibaba.fastjson2.JSON;
import com.yugao.fintech.antelope.infra.InfraApi;
import com.yugao.fintech.antelope.infra.module.group.model.rqrs.DataGroupBase;
import com.yugao.fintech.antelope.mps.api.constatnts.*;
import com.yugao.fintech.antelope.mps.api.model.content.MsgContent;
import com.yugao.fintech.antelope.mps.api.model.rqrs.MsgPushParam;
import com.yugao.fintech.antelope.mps.core.PushTaskContext;
import com.yugao.fintech.antelope.mps.core.parser.AbstractTemplateParser;
import com.yugao.fintech.antelope.mps.core.sender.MsgSendManager;
import com.yugao.fintech.antelope.mps.core.sender.MsgSender;
import com.yugao.fintech.antelope.mps.manager.MsgHistoryManager;
import com.yugao.fintech.antelope.mps.manager.MsgTemplateManager;
import com.yugao.fintech.antelope.mps.manager.ReceiverSubscribeManager;
import com.yugao.fintech.antelope.mps.model.ReceiverConfig;
import com.yugao.fintech.antelope.mps.model.entity.MsgHistory;
import com.yugao.fintech.antelope.mps.model.entity.MsgPushTask;
import com.yugao.fintech.antelope.mps.model.entity.MsgTemplate;
import com.yugao.fintech.antelope.mps.model.entity.MsgReceiver;
import com.yugao.fintech.antelope.mps.service.MsgPushTaskService;
import com.yugao.fintech.framework.assistant.utils.StringUtils;
import com.yugao.fintech.framework.assistant.utils.exception.BizException;
import com.yugao.fintech.framework.cache.core.service.CacheService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Service
public class MsgPushProcessImpl implements MsgPushProcess {
    @Autowired
    private MsgHistoryManager historyManager;
    @Autowired
    private MsgSendManager msgSendManager;
    @Autowired
    private CacheService cacheService;
    @Autowired
    private MsgPushTaskService pushTaskService;
    @Autowired
    private InfraApi infraApi;
    @Autowired
    private MsgTemplateManager templateManager;
    @Autowired
    private ReceiverSubscribeManager receiverSubscribeManager;


    @Override
    public void push(MsgPushParam req, String msgId) throws RuntimeException {
        PushTaskContext context = new PushTaskContext();
        // 解析消息内容
        MsgContent msgContent = req.getMsgContent();
        context.setMsgContent(msgContent);

        // 通过任务编码获取任务
        MsgPushTask pushTask = pushTaskService.getPushTaskByCode(msgContent.getTaskCode());
        context.setPushTask(pushTask);

        Set<String> toUserList = new HashSet<>();

        // 处理当前任务的接收者类型为通知管理员
        if (Objects.equals(pushTask.getReceiverType(), MsgCons.ReceiverType.ADMIN)) {
            List<MsgReceiver> msgReceiverList = receiverSubscribeManager.listTaskSubscriber(pushTask.getId());
            Set<String> toAdminUserList = msgReceiverList.stream().flatMap(e -> ReceiverConfig.of(e).stream())
                    .filter(e -> Objects.equals(e.getChannelType().getCode(), req.getChannelType()))
                    .map(ReceiverConfig::getToUser).collect(Collectors.toSet());
            if (StringUtils.isEmpty(toAdminUserList)) {
                log.warn("推送的任务 [{}] 是管理员通知类型, 但是没有任何接收者验证接受渠道信息", pushTask.getTaskCode());
                return;
            }
            toUserList.addAll(toAdminUserList);
        } else {
            if (StringUtils.isEmpty(msgContent.getToUser())) {
                throw new BizException("当前任务非管理员推送类型, 需要调用方指定接收者");
            }
            toUserList.add(msgContent.getToUser());
        }


        // 获取任务关联的模板并发送消息
        List<MsgTemplate> templateList = templateManager.listTemplateByTaskId(pushTask.getId(), req.getChannelType());
        log.info("taskCode: {}, templateCount: {}", msgContent.getTaskCode(), templateList.size());
        for (MsgTemplate template : templateList) {
            MsgHistory history = new MsgHistory();
            String groupName = null;

            // 获取标题, 优先级依次是 MsgContent.title, MsgTemplate.title, MsgPushTask.name
            msgContent.setTitle(StringUtils.priorityFirstNotEmpty(msgContent.getTitle(), template.getTitle(), pushTask.getName()));

            for (String toUser : toUserList) {
                try {
                    history.setTaskId(pushTask.getId()).setPushTime(new Date())
                            .setGroupId(pushTask.getGroupId()).setToUser(toUser);
                    context.getMsgContent().setToUser(toUser);
                    ChannelTypeEnum channelType = ChannelTypeEnum.ofByCode(template.getChannelType());
                    if (Objects.isNull(channelType)) {
                        log.warn("未查找到 channelType: {} 的渠道类型", template.getChannelType());
                        continue;
                    }
                    history.setChannelType(channelType.getCode());
                    groupName = Optional.ofNullable(infraApi.getDataGroup(pushTask.getGroupId())).map(DataGroupBase::getName)
                            .orElse("未知");

                    context.setChannelType(channelType);

                    // 推送消息
                    MsgSender sendHandler = msgSendManager.getMsgSendHandler(channelType);

                    // 解析模板数据
                    AbstractTemplateParser tplConfig = AbstractTemplateParser.jsonToObj(context.getMsgContent().getTplDataList(), template, channelType);
                    tplConfig.evaluate();

                    // 默认采用传进来的url, 否则采用后台配置的
                    msgContent.setPageUrl(StringUtils.defaultIfEmpty(msgContent.getPageUrl(), tplConfig.getPageUrl()));

                    context.setTplConfig(tplConfig);
                    context.setPushStatus(MsgPushStatus.SUCCESS);
                    sendHandler.sendMessage(context);
                    if (Objects.isNull(context.getChannelCode())) {
                        throw new BizException("channelCode is null");
                    }

                    // 因为当前方法消息队列监听器也会调用, 所以要抛出异常, 触发重试
                    if (MsgPushStatus.FAIL.equals(context.getPushStatus())) {
                        log.error("push fail, error: {}", context.getPushResult());
                        throw new BizException("推送失败, 请联系管理员");
                    }
                } catch (Exception e) {
                    context.setPushResult(StringUtils.defaultIfEmpty(context.getPushResult(), e.getMessage()));
                    context.setPushStatus(MsgPushStatus.FAIL);
                    throw e;
                } finally {
                    try {
                        String channelCode = Optional.ofNullable(context.getChannelCode()).map(ChannelCodeEnum::getCode).orElse("");
                        history.setPushResult(context.getPushResult()).setId(req.getHistoryId()).setTemplateId(template.getId())
                                .setBizKey(msgContent.getBizKey()).setBizType(msgContent.getBizType())
                                .setStatus(context.getPushStatus().getStatus()).setGroupName(groupName)
                                .setChannelCode(channelCode).setOriginalTemplateVar(context.getMsgContent().getTplDataList());
                        save(history, msgId);
                    } catch (Exception e) {
                        if (e instanceof NullPointerException) {
                            log.error("error: ", e);
                        }
                    } finally {
                        if (context.getPushStatus() == MsgPushStatus.FAIL) {
                            log.error("保存推送记录失败, content: {}", JSON.toJSONString(context));
                        }
                    }
                }
            }

        }

    }

    /**
     * 防止消费失败, 多次重试往数据库中插入重复数据, 如果多次重试之后, 依旧失败只保留最后一次的失败信息, 如果经过
     * 几次重试之后, 消费成功, 则保留成功的信息
     */
    private void save(MsgHistory history, String msgId) {
        if (Objects.nonNull(history.getId())) {
            MsgHistory entity = historyManager.getById(history.getId());
            history.setRetryCount(entity.getRetryCount() + 1);
            historyManager.updateById(history);
            return;
        }

        String key = MsgPushCacheKey.MSG_PUSH_SAVE.formatKey(msgId);
        Object value = cacheService.opsForValue().get(key);
        if (Objects.nonNull(value)) {
            history.setId(Long.parseLong(String.valueOf(value)));
            MsgHistory po = historyManager.getById(history.getId());
            history.setRetryCount(po.getRetryCount() + 1);
            historyManager.updateById(history);
            return;
        }
        historyManager.save(history);
        cacheService.opsForValue().set(key, history.getId());
    }
}
